package hbec.kettle.etl.plugins.common;

import java.sql.Connection;
import java.sql.ResultSet;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.common.exception.ExceptionTracker;
import hbec.kettle.etl.plugins.reader.mysqlreader.ParamKey;

public class DALUtils {
	private static final Logger logger = LoggerFactory.getLogger(DALUtils.class);
	
	public static Map<String, WatchEntity> getWatchSum(DataBase db, String table, Long position) {
		Map<String, WatchEntity> res = new HashMap<>();
		
		Set<String> temp = new HashSet<>();
		Connection conn = null;
		ResultSet rs = null;
		try {
			conn = DBSource.getConnection(db);
			rs = DBUtils.query(conn, MessageFormat.format(ParamKey.watchSumTableSql, table, String.valueOf(position)));
			if (rs != null && rs.isBeforeFirst()) {
				WatchEntity we = null;
				Map<Integer, Set<String>> identities = null;
				Map<Integer, Set<Long>> ids = null;
				Long minPosition = 0L;
				Long maxPosition = 0L;
				while (rs.next()) {
					if (res.containsKey(rs.getString("object"))) {
						we = res.get(rs.getString("object"));
						identities = we.getIdentities();
						ids = we.getIds();
						maxPosition = rs.getLong("id");
					} else { 
						we = new WatchEntity();
						identities = new HashMap<>();
						ids = new HashMap<>();
						we.setIdentities(identities);
						we.setIds(ids);
						minPosition = rs.getLong("id");
					}
					we.setObject(rs.getString("object"));
					we.setMinPosition(minPosition);
					we.setMaxPosition(maxPosition);
					
					Integer actionId = 0;
					switch (rs.getString("action")) {
						case "INSERT":
							actionId = Integer.valueOf(1);
							break;
						case "UPDATE":
							actionId = Integer.valueOf(2);
							break;
						case "DELETE":
							actionId = Integer.valueOf(4);
							break;
						default:
							break;
					}
					if (identities.containsKey(actionId)) { 
						if (temp.contains(rs.getString("identity"))) {
							for (Integer k : identities.keySet()) {
								identities.get(k).remove(rs.getString("identity"));
							} 
						}  
						identities.get(actionId).add(rs.getString("identity"));
						
						ids.get(actionId).add(rs.getLong("id"));
					} else {
						Set<String> _identities = new HashSet<>();
						Set<Long> _ids = new HashSet<>();
						_identities.add(rs.getString("identity"));
						_ids.add(rs.getLong("id"));
						identities.put(actionId, _identities);
						ids.put(actionId, _ids);
					}
					we.setTimestamp(rs.getTimestamp("timestamp").getTime());
					
					res.put(rs.getString("object"), we);
					temp.add(rs.getString("identity"));
				}
			}
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeResultSet(rs);
			DBUtils.closeConnection(conn);
		}
		
		return res;
	} 
	
	public static Map<String, Long> getCheckPoint(DataBase db, String table) {
		Map<String, Long> res = new HashMap<>();
		
		Connection conn = null;
		ResultSet rs = null;
		try {
			conn = DBSource.getConnection(db);
			rs = DBUtils.query(conn, MessageFormat.format(ParamKey.checkpointTableSql, table));
			if (rs != null && rs.isBeforeFirst()) {
				while (rs.next()) {
					res.put(rs.getString(0), rs.getLong(1));
				}
			}
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeResultSet(rs);
			DBUtils.closeConnection(conn);
		}
		
		return res;
	}
	
	public static int setCheckPoint(DataBase db, String table, String object, Long posistion) {
		int res = 0;
		
		Connection conn = null;
		try {
			conn = DBSource.getConnection(db);
			res = DBUtils.update(conn, MessageFormat.format(ParamKey.updateCheckpointTableSql, table, object, String.valueOf(posistion)));
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeConnection(conn);
		}
		
		return res;
	}
	
	public static int initializeSqlite(DataBase db, String table) {
		int res = 0;
		
		Connection conn = null;
		try {
			conn = DBSource.getConnection(db);
			res = DBUtils.update(conn, MessageFormat.format(ParamKey.dropSqliteTableSql, table));
			if (res > 0) {
				res = DBUtils.update(conn, MessageFormat.format(ParamKey.createSqliteTableSql, table));
			}
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeConnection(conn);
		}
		
		return res;
	}
	
	public static int setSqlite(DataBase db, String table, String object, String line) {
		int res = 0;
		
		Connection conn = null;
		try {
			conn = DBSource.getConnection(db);
			res = DBUtils.update(conn, MessageFormat.format(ParamKey.insertSqliteTableSql, table, object, line));
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeConnection(conn);
		}
		
		return res;
	}
	
	public static List<WatchLine> getSqlite(DataBase db, String table, String object, Long id) {
		List<WatchLine> res = new ArrayList<>();
		
		Connection conn = null;
		ResultSet rs = null;
		try {
			conn = DBSource.getConnection(db);
			rs = DBUtils.query(conn, MessageFormat.format(ParamKey.querySqliteTableSql, table, object, id));
			if (rs != null && rs.isBeforeFirst()) {
				while (rs.next()) {
					WatchLine wl = new WatchLine();
					wl.setId(rs.getLong(0));
					wl.setLine(rs.getString(1));
					res.add(wl);
				}
			}
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			DBUtils.closeResultSet(rs);
			DBUtils.closeConnection(conn);
		}
		
		return res;
	}
}
